package com.bw.medical.realtime.dwd.app;

import com.bw.medical.realtime.common.base.BaseSQL;
import com.bw.medical.realtime.common.constant.Constant;
import com.bw.medical.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionReview extends BaseSQL {
    public static void main(String[] args) {
        new DwdInteractionReview().start(
                10018,
                4,
                "medical_dwd_interaction_review"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsDb(tabEnv,"medical_dwd_interaction_review");

        Table tables = tabEnv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['update_time'] review_time,\n" +
                "`data`['rating'] rating,\n" +
                "`data`['doctor_id'] doctor_id,\n" +
                "`data`['patient_id'] patient_id,\n" +
                "`data`['user_id'] user_id\n" +
                "from \n" +
                "topic_db\n" +
                "where `table` = 'consultation'\n" +
                "and `type` = 'update'\n" +
                "and `data`['status'] = '207'");

        tabEnv.createTemporaryView("interaction_table",tables);

//       tables.execute().print();

        tabEnv.executeSql("create table " + Constant.DWD_TRADE_INTERACTION_REVIEW + "(\n" +
                "`id` string,\n" +
                "`review_time` string,\n" +
                "`rating` string,\n" +
                "`doctor_id` string,\n" +
                "`patient_id` string,\n" +
                "`user_id` string" +
                ") " + SQLUtil.getKafkaSinkSQL(Constant.DWD_TRADE_INTERACTION_REVIEW ));

        tables.executeInsert(Constant.DWD_TRADE_INTERACTION_REVIEW);

    }
}
